package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {

    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {

        KStream<String, String> kStream = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });

        KGroupedStream<String, String> kGroupedStream = kStream.groupByKey();

        //根据时间窗口统计
        TimeWindowedKStream<String, String> windowedKStream = kGroupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(60)));

        //根据每个组统计结果
        // 0 阅读 1 点赞 2 评论 3 收藏
        //{"articleId":1471738975990321153,"view":1,"collect":3,"comment":0,"like":1}
        //{"articleId":1503210511334916097,"view":1,"collect":3,"comment":0,"like":1}

        //初始化器  为空
        Initializer<String> initializer = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };

        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                //value 对应的是 UpdateArticleMessage实体类的json对象
                //aggregate 对应的是 ArticleStreamMessage 实体类的json对象

                ArticleStreamMessage articleStreamMessage = null;

                if (aggregate == null) {
                    articleStreamMessage = new ArticleStreamMessage();
                    articleStreamMessage.setArticleId(Long.parseLong(key));
                    articleStreamMessage.setView(0);
                    articleStreamMessage.setCollect(0);
                    articleStreamMessage.setComment(0);
                    articleStreamMessage.setLike(0);
                } else {
                    articleStreamMessage = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }

                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Integer type = updateArticleMessage.getType();

                // 0 阅读 1 点赞 2 评论 3 收藏
                switch (type) {
                    case 0: {
                        articleStreamMessage.setView(articleStreamMessage.getView() + updateArticleMessage.getAdd());
                        break;
                    }
                    case 1: {
                        articleStreamMessage.setLike(articleStreamMessage.getLike() + updateArticleMessage.getAdd());
                        break;
                    }
                    case 2: {
                        articleStreamMessage.setComment(articleStreamMessage.getComment() + updateArticleMessage.getAdd());
                        break;
                    }
                    case 3: {
                        articleStreamMessage.setCollect(articleStreamMessage.getCollect() + updateArticleMessage.getAdd());
                        break;
                    }
                }


                return JSON.toJSONString(articleStreamMessage);
            }
        };
        KTable<Windowed<String>, String> kTable = windowedKStream.aggregate(initializer, aggregator);

        KStream<String, String> kStream1 = kTable.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });

        //把上面的结果放到另一个topic中
        return kStream1;
    }

}
